package bjut.zrpc.core.optimization.netty;


import bjut.zrpc.core.optimization.constant.ZRpcConstant;
import bjut.zrpc.core.optimization.enums.ZRpcCompressEnum;
import bjut.zrpc.core.optimization.enums.ZRpcSerializerEnum;
import bjut.zrpc.core.optimization.factory.SingletonFactory;
import bjut.zrpc.core.optimization.message.ZRpcMessage;
import bjut.zrpc.core.optimization.message.ZRpcResponseMessage;
import bjut.zrpc.core.optimization.processor.UnprocessedRequest;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;


import java.net.InetSocketAddress;

/**
 * @Author: zhuzhiming
 * @Date: 2022/6/24 14:43
 */
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    private final UnprocessedRequest unprocessedRequest;
    private final NettyClient nettyClient;

    public NettyClientHandler() {
        unprocessedRequest = SingletonFactory.getInstance(UnprocessedRequest.class);
        nettyClient = SingletonFactory.getInstance(NettyClient.class);
    }


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {


        try {
            if (msg instanceof ZRpcMessage) {

                ZRpcMessage zRpcMessage = (ZRpcMessage) msg;

                byte messageType = zRpcMessage.getMessageType();

                if (messageType == ZRpcConstant.RESPONSE_TYPE) {

                    ZRpcResponseMessage<Object> responseMessage = (ZRpcResponseMessage<Object>) zRpcMessage.getBodyData();
                    unprocessedRequest.complete(responseMessage);
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage());
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

        if (evt instanceof IdleStateEvent) {
            if (((IdleStateEvent) evt).state() == IdleState.WRITER_IDLE) {

                ZRpcMessage zRpcMessage = ZRpcMessage.builder()
                        .compress(ZRpcCompressEnum.GZIP.getType())
                        .codec(ZRpcSerializerEnum.PROTOSTUFF.getCodec())
                        .messageType(ZRpcConstant.HEART_REQUEST_TYPE)
                        .bodyData(ZRpcConstant.HEART_REQUEST_DATA)
                        .build();

                Channel channel = nettyClient.getChannel((InetSocketAddress) ctx.channel().remoteAddress());
                channel.writeAndFlush(zRpcMessage);

            }

        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        ctx.channel().close();
    }
}
